package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.commont.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/4/26 09:02
 */
public class Dwd_07_DwdTradeOrderRefund extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_07_DwdTradeOrderRefund().init(
            30007,
            2,
            "Dwd_07_DwdTradeOrderRefund"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_07_DwdTradeOrderRefund");
        // 读取字典表
        readBaseDic(tEnv);
        
        // 2. 过滤退单表数据
        Table orderRefundInfo = tEnv.sqlQuery(
            "select " +
                "data['id'] id,  " +
                "data['user_id'] user_id,  " +
                "data['order_id'] order_id,  " +
                "data['sku_id'] sku_id,  " +
                "data['refund_type'] refund_type,  " +
                "data['refund_num'] refund_num,  " +
                "data['refund_amount'] refund_amount,  " +
                "data['refund_reason_type'] refund_reason_type,  " +
                "data['refund_reason_txt'] refund_reason_txt,  " +
                "data['create_time'] create_time,  " +
                "pt,  " +
                "ts  " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='order_refund_info' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);
        // 3. 过滤出订单表数据
        Table orderInfo = tEnv.sqlQuery(
            "select " +
                "data['id'] id,  " +
                "data['province_id'] province_id,  " +
                "`old`  " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='order_info' " +
                "and `type`='update' " +
                "and `old`['order_status'] is not null " +
                "and `data`['order_status']='1005' ");
        tEnv.createTemporaryView("order_info", orderInfo);
        // 4. 3 张表的 join
        Table result = tEnv.sqlQuery(
            "select " +
                "ori.id,  " +
                "ori.user_id,  " +
                "ori.order_id,  " +
                "ori.sku_id,  " +
                "oi.province_id,  " +
                "date_format(ori.create_time,'yyyy-MM-dd') date_id,  " +
                "ori.create_time,  " +
                "ori.refund_type refund_type_code,  " +
                "dic1.dic_name refund_type_name,  " +
                "ori.refund_reason_type refund_reason_type_code,  " +
                "dic2.dic_name refund_reason_type_name,  " +
                "ori.refund_reason_txt,  " +
                "ori.refund_num,  " +
                "ori.refund_amount,  " +
                "ori.ts  " +
                "from order_refund_info ori " +
                "join order_info oi " +
                "on ori.order_id=oi.id " +
                "join base_dic for system_time as of ori.pt as dic1 " +
                "on ori.refund_type=dic1.dic_code " +
                "join base_dic for system_time as of ori.pt as dic2 " +
                "on ori.refund_reason_type=dic2.dic_code ");
        
        // 5. join 的结果写出到 kafka 中
        tEnv.executeSql(
            "create table dwd_trade_order_refund(  " +
                "id string,  " +
                "user_id string,  " +
                "order_id string,  " +
                "sku_id string,  " +
                "province_id string,  " +
                "date_id string,  " +
                "create_time string,  " +
                "refund_type_code string,  " +
                "refund_type_name string,  " +
                "refund_reason_type_code string,  " +
                "refund_reason_type_name string,  " +
                "refund_reason_txt string,  " +
                "refund_num string,  " +
                "refund_amount string,  " +
                "ts bigint  " +
                ")" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));
        result.executeInsert("dwd_trade_order_refund");
    }
}
/*
交易域退单事务事实表
    数据源:
        退单表: order_refund_info      insert
        
        订单表: order_info     update
        
        字典表:  base_dic

 */